#include "common.h"
#include "config.h"
#include "cuckoo.h"
#include "arraylist.h"

#define TEST_NZ(x) do { if ( (x)) die("error: " #x " failed (returned non-zero)." ); } while (0)
#define TEST_Z(x)  do { if (!(x)) die("error: " #x " failed (returned zero/null)."); } while (0)

DataNode* datanodes=NULL;
int nodeid;
struct Context *s_ctx = NULL;
Connection** datanodes_conns_as_server=NULL;
Connection** datanodes_conns_as_client=NULL;
struct rdma_event_channel *event_channel = NULL;
int flag=0;
long long id=0;
cuckoo* update_ctx_map=NULL;
char update_block[BLK_SIZE];
int parity[BLK_SIZE];
Array_List* log_list = NULL;
pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond=PTHREAD_COND_INITIALIZER;

//================================================================================
void connect2datanodes();
void* thread4connect2datanodes();
int on_connect_request(struct rdma_cm_id *id);
int on_connection(struct rdma_cm_id *id);
int on_disconnect(struct rdma_cm_id *id);
void destroy_connection(void *context);
int on_addr_resolved(struct rdma_cm_id *id);
int on_route_resolved(struct rdma_cm_id *id);
int on_event(struct rdma_cm_event *event);
void build_context(struct ibv_context *verbs);
void build_qp_attr(struct ibv_qp_init_attr *qp_attr);
void register_memory(Connection *conn);
void build_params(struct rdma_conn_param *params);
void post_receives(struct Connection *conn);
void send_MR(int fd,MemRegionMessage* mr_message);
void recv_MR(int fd,MemRegionMessage* mr_message);
void* poll_cq(void *ctx);
void on_completion(struct ibv_wc *wc);
void send_WriteResponse(WriteResponse* response,Connection* conn);
void send_ReadResponse(ReadResponse* response,Connection* conn);
void send_UpdateEntity(UpdateEntity* entity,int paritynodeid,long long id);
void send_UpdateMessage(int fd,UpdateMessage* message);
void recv_UpdateMessage(int fd,UpdateMessage* message);

void* thread4accpetupdate(void* arg);
void* thread4processupdate(void* arg);
void* thread4processloglist(void* arg);
void process_memlog(char* memlog);
void computeAndUpdate(cuckoo* init_block_map,cuckoo* latest_block_map);

int main(int argc, char const *argv[])
{
    update_ctx_map = cuckoo_init(100);
    log_list = Array_List_Init();
    datanodes = read_config_Node(NODE_CONFIG);
	  nodeid = current_node(datanodes);
    datanodes_conns_as_server = (Connection**)malloc(sizeof(Connection*)*DATANODE_NUM);
    datanodes_conns_as_client = (Connection**)malloc(sizeof(Connection*)*DATANODE_NUM);


    pthread_t processloglist_threadid;
    pthread_create(&processloglist_threadid,NULL,thread4processloglist,NULL);

    pthread_t accpet_client4update_threadid;
    pthread_create(&accpet_client4update_threadid,NULL,thread4accpetupdate,NULL);

    //创建连接线程
    pthread_t connect2datanodes_threadid;
    pthread_create(&connect2datanodes_threadid,NULL,thread4connect2datanodes,NULL);

	  struct sockaddr_in addr;
	  struct rdma_cm_event *event = NULL;
  	struct rdma_cm_id *listener_id = NULL;	
  	uint16_t port = RDMA_PORT;
  	memset(&addr, 0, sizeof(addr));
  	addr.sin_family = AF_INET;
  	addr.sin_port = htons(port);
  	addr.sin_addr.s_addr = htonl(INADDR_ANY);

  	TEST_Z(event_channel=rdma_create_event_channel());
  	TEST_NZ(rdma_create_id(event_channel,&listener_id,NULL,RDMA_PS_TCP));
  	TEST_NZ(rdma_bind_addr(listener_id,(struct sockaddr *)&addr));
  	TEST_NZ(rdma_listen(listener_id,10));//backlog=10

  	printf("datanode[%s:%d] is listening now,\n",datanodes[nodeid].rdma_ip,port);
  	printf("current_node id is %d\n",nodeid);

  	while (rdma_get_cm_event(event_channel, &event) == 0) {
	    struct rdma_cm_event event_copy;
	    memcpy(&event_copy, event, sizeof(*event));
	    rdma_ack_cm_event(event);
	    if (on_event(&event_copy))
	    	break;
  	}

    rdma_destroy_id(listener_id);
    rdma_destroy_event_channel(event_channel);

    pthread_cancel(connect2datanodes_threadid);
    pthread_cancel(processloglist_threadid);
    pthread_cancel(accpet_client4update_threadid);
	  return 0;
}

void* thread4accpetupdate(void* arg){
  //pthread_detach(pthread_self());
  int listen_fd = server_bind(UPDATE_SOCKET_PORT,10);
  while(1){
        int conn_fd = server_accept(listen_fd);
        pthread_t threadid;
        pthread_create(&threadid,NULL,thread4processupdate,(void*)conn_fd);
  }
  return NULL;
}

int checkMemorylog(char* memlog){
    int offset = 0;
    int size = sizeof(UpdateEntity);
    UpdateEntity* entity = (UpdateEntity*)malloc(sizeof(UpdateEntity));
    while(1){
      if (offset+size>RDMA_BUFFER_SIZE) break;
      memcpy(entity,memlog+offset,size);
      offset=offset+size;
      if (entity->checksum!=CHECK_SUM) break;
    }
    free(entity);
    return offset;
}

void* thread4processupdate(void* arg){
    printf("begin to process update\n");
    pthread_detach(pthread_self());
    int fd = (int)arg;
    //接受update请求，复制一份log加入到list中，原来的log清零，注意这里和log_list处理线程构成一生产者多消费者模式
    UpdateMessage* message = (UpdateMessage*)malloc(sizeof(UpdateMessage));
    recv_UpdateMessage(fd,message);

    Connection* conn = datanodes_conns_as_server[message->target_region];
    DataNode_Connection* datanode_conn = (DataNode_Connection*)conn->conn_info;

    char* memlog = (char*)malloc(RDMA_BUFFER_SIZE);
    memcpy(memlog,datanode_conn->rdma_region,RDMA_BUFFER_SIZE);
    memset(datanode_conn->rdma_region,'\0',RDMA_BUFFER_SIZE);

    printf("messagetype:%d\n",message->type);
    if (message->type==2)
    {
        printf("123123\n");
        //检查memorylog完整性
        int v_len = checkMemorylog(memlog);
        message->v_len = v_len;
        send_UpdateMessage(fd,message);
        recv_UpdateMessage(fd,message);
        if (v_len!=message->v_len)
        {
            memset(memlog+message->v_len,'\0',RDMA_BUFFER_SIZE-message->v_len);
        }
    }

    //将memlog加入到log_list中
    //note 加锁，实现1个生产者
    pthread_mutex_lock(&mutex);

    if (Array_List_IsEmpty(log_list)==1)
    {
      pthread_cond_signal(&cond);
    }
    Array_List_Insert(log_list,(void*)memlog,-1);

    pthread_mutex_unlock(&mutex);

    if (message->type==2)
    {
        while(1){
          if (Array_List_IsEmpty(log_list)==1){
            break;
          }
        }
        message->type=3; 
    }else{
        message->type=1;
    }
    send_UpdateMessage(fd,message);
    free(message);
    close(fd);
    printf("end process update\n");
    return NULL;
}

void* thread4processloglist(void* arg){
    char* tmp=NULL;
    while(1){
        pthread_mutex_lock(&mutex);
        if (Array_List_IsEmpty(log_list)==1)
        {
          pthread_cond_wait(&cond,&mutex);
        }
        tmp = (char*)Array_List_GetAt(log_list,0);
        Array_List_RemoveAt(log_list,0);
        pthread_mutex_unlock(&mutex);

        process_memlog(tmp);
    }
    return NULL;
}

void process_memlog(char* memlog){
  //遍历解析memlog，反序列化为UpdateEntity，检查checksum，通过map存储，最后根据blockid更新数据
  printf("start process_memlog\n");
  int offset = 0;
  int size = sizeof(UpdateEntity);
  int int_size = sizeof(int);
  UpdateEntity* entity = (UpdateEntity*)malloc(sizeof(UpdateEntity));

  cuckoo* init_block_map = cuckoo_init(100);
  cuckoo* latest_block_map = cuckoo_init(100);

  while(1){
    if (offset+size>RDMA_BUFFER_SIZE) break;
    memcpy(entity,memlog+offset,size);
    if (entity->checksum!=CHECK_SUM) break;
    int blockid = entity->blockid;
    int parityid = entity->parityid;
    char* data = entity->newdata;

    char* key = convert_int_to_string(blockid);

    if (cuckoo_get(init_block_map,key)==NULL){
        cuckoo_insert(init_block_map,key,(void*)(memlog+offset+int_size*2));
        cuckoo_insert(latest_block_map,key,(void*)(memlog+offset+int_size*2));
    }else{
        cuckoo_remove(latest_block_map,key);
        cuckoo_insert(latest_block_map,key,(void*)(memlog+offset+int_size*2));
    }

    offset=offset+size;
    free(key);
  }

  //遍历map，更新
  computeAndUpdate(init_block_map,latest_block_map);

  cuckoo_destroy(init_block_map,0);
  cuckoo_destroy(latest_block_map,0);
  free(entity);
  free(memlog);
  printf("end process_memlog\n");
}

void computeAndUpdate(cuckoo* init_block_map,cuckoo* latest_block_map){
  printf("start computeAndUpdate\n");
  for (int i = 0; i < init_block_map->cap; i++)
    {
        if (init_block_map->nodes[i].taken)
        {
            char* old = (char*)init_block_map->nodes[i].val;
            char* new = (char*)cuckoo_get(latest_block_map,init_block_map->nodes[i].key);
            for (int j = 0; j < BLK_SIZE; ++j) parity[j] = parity[j]+new[j]-old[j];
        }
    }
  printf("end computeAndUpdate\n");
}

//连接其他datanodes 用于rdma-write
void connect2datanodes(){
  datanodes_conns_as_server[nodeid]=NULL;

  for (int i = 0; i < DATANODE_NUM; ++i)
  {
    flag=0;
    if (i!=nodeid)
    {
      char* target_ip=datanodes[i].rdma_ip;
      //与目标datanode建立连接，并且将conn存入datanodes_conns_as_server
      //note:不可与自己连接，故数组中有nodeid坐标元素为NULL
      struct rdma_cm_id *conn_id= NULL;
      // struct sockaddr_in *dst_addr = (struct sockaddr_in*)malloc(sizeof(struct sockaddr_in));
      // dst_addr->sin_family=AF_INET;
      // dst_addr->sin_port = htons(RDMA_PORT);
      // dst_addr->sin_addr.s_addr = inet_addr(target_ip);
      struct addrinfo *addr;
      char port[20];
      sprintf(port,"%d",RDMA_PORT);
      TEST_NZ(getaddrinfo(target_ip, port,NULL,&addr));
      TEST_NZ(rdma_create_id(event_channel, &conn_id, NULL, RDMA_PS_TCP));
      TEST_NZ(rdma_resolve_addr(conn_id, NULL, addr->ai_addr, TIMEOUT_IN_MS));
      printf("%s\n", target_ip);
      while(flag==0){};
    }
  }
}

void* thread4connect2datanodes(void* arg){
    sleep(10);
    connect2datanodes();
}

int on_connect_request(struct rdma_cm_id *id)
{
  printf("received connection request.\n");

  //根据id判断来自client连接或是来自datanode连接
  struct sockaddr* addr =  rdma_get_peer_addr(id);
  int peer_id = get_nodeid(addr,datanodes);
  printf("the connection request comes from:%d\n", peer_id);

  //上下文,保护域,完成队列
  build_context(id->verbs);
  //创建QP列队,设置QP参数
  struct ibv_qp_init_attr qp_attr;
  build_qp_attr(&qp_attr);
  //qp连接
  TEST_NZ(rdma_create_qp(id, s_ctx->pd, &qp_attr));

  Connection* conn = (Connection*)malloc(sizeof(Connection));
  if (peer_id==-1)//client
  {
    conn->type = 0;
    Client_Connection* client_conn = (Client_Connection*)malloc(sizeof(Client_Connection));
    conn->conn_info = (void*)client_conn;

  }else{//datanode
    conn->type = 1;
    DataNode_Connection* datanode_conn = (DataNode_Connection*)malloc(sizeof(DataNode_Connection));
    conn->conn_info = (void*)datanode_conn;
    datanodes_conns_as_server[peer_id] = conn;//将连接的datanode连接信息按照nodeid保存
  }
  conn->qp = id->qp;
  conn->id = id;
  conn->len=0;
  id->context = conn;
  //注册内存区域
  register_memory(conn);
  
  struct rdma_conn_param cm_params;
  //创建连接参数  
  build_params(&cm_params);
  //接受来自client或者datanode的连接
  TEST_NZ(rdma_accept(id, &cm_params));

  if (conn->type==0)//client
  {
    //发送一个receive消息
    post_receives(conn);
  }

  return 0;
}

void* thread4MR(void* arg){
  pthread_detach(pthread_self());
  struct rdma_cm_id* id = (struct rdma_cm_id*)arg;
  Connection* conn = (Connection*)id->context;
  struct sockaddr* addr =  rdma_get_peer_addr(id);
  int peer_id = get_nodeid(addr,datanodes);
  if (conn->type==1)
  {
    //通过socket传递 mr给datanode
    int listen_fd = server_bind(REGION_SOCKET_PORT+peer_id+1,10);
    int fd = server_accept(listen_fd);
    MemRegionMessage* mr_message = (MemRegionMessage*)malloc(sizeof(MemRegionMessage));
    DataNode_Connection* datanode_conn = (DataNode_Connection*)conn->conn_info;
    memcpy(&mr_message->mr, datanode_conn->rdma_mr, sizeof(struct ibv_mr));
    send_MR(fd,mr_message);
    free(mr_message);
    close(fd);
    close(listen_fd);
    printf("accept datanode connection successfully\n");
  }else if (conn->type==2)
  {
     //请求并接受datanode_server的remote_mr
    int fd = connect_try(datanodes[peer_id].tcp_ip,REGION_SOCKET_PORT+nodeid+1);
    MemRegionMessage* mr_message = (MemRegionMessage*)malloc(sizeof(MemRegionMessage));
    recv_MR(fd,mr_message);
    DataNode_Connection* datanode_conn = (DataNode_Connection*)conn->conn_info;
    memcpy(&datanode_conn->peer_mr,&mr_message->mr,sizeof(struct ibv_mr));
    free(mr_message);
    close(fd);
    printf("connection to datanode successfully\n");
  }else{
    printf("unknown connection type\n");
  }

  flag=1;

  return NULL;
}

int on_connection(struct rdma_cm_id *id)
{
  Connection* conn = (Connection*)id->context;
  if (conn->type==0)
  {
    printf("accept client connection successfully\n");
  }else if (conn->type==1||conn->type==2)
  {
    pthread_t thread4MRid;
    pthread_create(&thread4MRid,NULL,thread4MR,(void*)id);
  }else{
    printf("error connection type\n");
    return -1;
  }
  return 0;
}

int on_disconnect(struct rdma_cm_id *id)
{
  printf("peer disconnected.\n");
  destroy_connection(id->context); 
  return 0;
}

void destroy_connection(void *context)
{
  struct Connection *conn = (struct Connection *)context;
  rdma_destroy_qp(conn->id);

  if (conn->type==0)//client
  {
    Client_Connection* client_conn = (Client_Connection*)conn->conn_info;
    ibv_dereg_mr(client_conn->send_mr);
    ibv_dereg_mr(client_conn->recv_mr);
    free(client_conn->recv_region);
    free(client_conn->send_region);
  }else if(conn->type==1){//datanode
    DataNode_Connection* datanode_conn = (DataNode_Connection*)conn->conn_info;
    ibv_dereg_mr(datanode_conn->rdma_mr);
    free(datanode_conn->rdma_region);
  }else{
    DataNode_Connection* datanode_conn = (DataNode_Connection*)conn->conn_info;
    ibv_dereg_mr(datanode_conn->rdma_mr);
    free(datanode_conn->rdma_region);
  }
  rdma_destroy_id(conn->id);
  free(conn);
}

int on_addr_resolved(struct rdma_cm_id *id)
{
  printf("start address resolved.\n");
  struct sockaddr* addr =  rdma_get_peer_addr(id);

  int peer_id = get_nodeid(addr,datanodes);

  printf("on_addr_resolved:%d\n",peer_id);

  struct ibv_qp_init_attr qp_attr;
  build_context(id->verbs);
  build_qp_attr(&qp_attr);
  TEST_NZ(rdma_create_qp(id, s_ctx->pd, &qp_attr));

  Connection* conn = (Connection*)malloc(sizeof(Connection));
  conn->type=2;
  conn->qp = id->qp;
  conn->id = id;
  conn->len=0;
  id->context = conn;
  datanodes_conns_as_client[peer_id] = conn;
  DataNode_Connection* datanode_conn = (DataNode_Connection*)malloc(sizeof(DataNode_Connection));
  conn->conn_info = (void*)datanode_conn;

  register_memory(conn);
  TEST_NZ(rdma_resolve_route(id, TIMEOUT_IN_MS));
  printf("end address resolved.\n");
  return 0;
}

int on_route_resolved(struct rdma_cm_id *id)
{
  struct rdma_conn_param cm_params;
  printf("route resolved.\n");
  build_params(&cm_params);
  TEST_NZ(rdma_connect(id, &cm_params));
  return 0;
}

int on_event(struct rdma_cm_event *event)
{
  int r = 0;
  printf("%s\n",rdma_event_str(event->event));
  if (event->event == RDMA_CM_EVENT_ADDR_RESOLVED)
    r = on_addr_resolved(event->id);
  else if (event->event == RDMA_CM_EVENT_ROUTE_RESOLVED)
    r = on_route_resolved(event->id);
  else if (event->event == RDMA_CM_EVENT_CONNECT_REQUEST)
    r = on_connect_request(event->id);
  else if (event->event == RDMA_CM_EVENT_ESTABLISHED)
    r = on_connection(event->id);
  else if (event->event == RDMA_CM_EVENT_DISCONNECTED)
    r = on_disconnect(event->id);
  else{
    die("on_event: unknown event.");
    //printf("unknown event.\n");
  }
  return r;
}


void build_context(struct ibv_context *verbs)
{
  if (s_ctx) {
    if (s_ctx->ctx != verbs)
      die("cannot handle events in more than one context.");

    return;
  }

  s_ctx = (struct Context *)malloc(sizeof(struct Context));

  s_ctx->ctx = verbs;

  TEST_Z(s_ctx->pd = ibv_alloc_pd(s_ctx->ctx));
  TEST_Z(s_ctx->comp_channel = ibv_create_comp_channel(s_ctx->ctx));
  TEST_Z(s_ctx->cq = ibv_create_cq(s_ctx->ctx,64, NULL, s_ctx->comp_channel, 0)); /* cqe=10 is arbitrary */
  TEST_NZ(ibv_req_notify_cq(s_ctx->cq, 0));

  TEST_NZ(pthread_create(&s_ctx->cq_poller_thread, NULL, poll_cq, NULL));
}

void build_qp_attr(struct ibv_qp_init_attr *qp_attr)
{
  memset(qp_attr, 0, sizeof(*qp_attr));

  qp_attr->send_cq = s_ctx->cq;
  qp_attr->recv_cq = s_ctx->cq;
  qp_attr->qp_type = IBV_QPT_RC;

  qp_attr->cap.max_send_wr = 64;
  qp_attr->cap.max_recv_wr = 64;
  qp_attr->cap.max_send_sge = 1;
  qp_attr->cap.max_recv_sge = 1;
}

void register_memory(Connection *conn)
{
  if (conn->type==0)//client
  {
    Client_Connection* client_conn = (Client_Connection*)conn->conn_info;
    client_conn->recv_region = (char*)malloc(SEND_RECV_BUFFER_SIZE);
    client_conn->send_region = (char*)malloc(SEND_RECV_BUFFER_SIZE);

    TEST_Z(client_conn->send_mr = ibv_reg_mr(
          s_ctx->pd,
          client_conn->send_region,
          SEND_RECV_BUFFER_SIZE,
          IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE));

    TEST_Z(client_conn->recv_mr = ibv_reg_mr(
          s_ctx->pd,
          client_conn->recv_region,
          SEND_RECV_BUFFER_SIZE,
          IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE));

  }else if (conn->type==1||conn->type==2)
  {//datanode
    DataNode_Connection* datanode_conn = (DataNode_Connection*)conn->conn_info;
    datanode_conn->rdma_region = (char*)malloc(RDMA_BUFFER_SIZE);

    TEST_Z(datanode_conn->rdma_mr = ibv_reg_mr(
          s_ctx->pd,
          datanode_conn->rdma_region,
          RDMA_BUFFER_SIZE,
          IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE));

  }else{
    printf("error type in register_memory");
    return;
  }
}


void build_params(struct rdma_conn_param *params)
{
  memset(params, 0, sizeof(*params));
  params->initiator_depth = params->responder_resources = 1;
  params->rnr_retry_count = 7; /* infinite retry */
}


void post_receives(struct Connection *conn)
{
  if (conn->type==0)
  {
    struct ibv_recv_wr wr, *bad_wr = NULL;
    struct ibv_sge sge;
    //准备工作请求
    wr.wr_id = (uintptr_t)conn;
    wr.next = NULL;
    wr.sg_list = &sge;
    wr.num_sge = 1;

    Client_Connection* client_conn = (Client_Connection*)conn->conn_info;

    sge.addr = (uintptr_t)client_conn->recv_region;
    sge.length = sizeof(struct WriteRequest);
    sge.lkey = client_conn->recv_mr->lkey;

    TEST_NZ(ibv_post_recv(conn->qp, &wr, &bad_wr));
  }else{
    printf("datanode cannot communicate with each other by send/recv\n");
    return;
  }
}

void send_MR(int fd,MemRegionMessage* mr_message)
{
  size_t needSend =sizeof(MemRegionMessage);
  char *sendBuf=(char *)malloc(needSend);
  memcpy(sendBuf,mr_message,needSend); 
  send_bytes(fd,sendBuf,needSend);
  free(sendBuf);
}

void recv_MR(int fd,MemRegionMessage* mr_message){
  size_t needRecv =sizeof(MemRegionMessage);
  char *recvBuf=(char *)malloc(needRecv);
  recv_bytes(fd,recvBuf,needRecv);
  memcpy(mr_message,recvBuf,needRecv);  //反序列化
  free(recvBuf);
}

void * poll_cq(void *ctx)
{
  struct ibv_cq *cq;
  struct ibv_wc wc;

  while (1) {
    TEST_NZ(ibv_get_cq_event(s_ctx->comp_channel, &cq, &ctx));
    ibv_ack_cq_events(cq, 1);
    TEST_NZ(ibv_req_notify_cq(cq, 0));

    while (ibv_poll_cq(cq, 1, &wc))
      on_completion(&wc);
  }

  return NULL;
}

//request的处理逻辑
void on_completion(struct ibv_wc *wc)
{
    printf("start sloving the wc\n");
    if (wc->status != IBV_WC_SUCCESS)
      die("on_completion: status is not IBV_WC_SUCCESS.");
    if (wc->opcode == IBV_WC_RECV){
      Connection* conn = (struct Connection *)(uintptr_t)wc->wr_id;
      post_receives(conn);
      Client_Connection* client_conn = (Client_Connection*)conn->conn_info;

      int trace_type=0;
      memcpy(&trace_type,client_conn->recv_region,sizeof(int));

      if (trace_type==0)
      {
          ReadRequest* request = (ReadRequest*)client_conn->recv_region;
          //memcpy(request,client_conn->recv_region,sizeof(ReadRequest));
          //解析ReadRequest，根据blockid找到数据，发送ReadResponse
          ReadResponse* response = (ReadResponse*)malloc(sizeof(ReadResponse));
          response->type=0;
          response->responseid = request->requestid;
          response->res = 1;
          gene_radm_buff(response->newdata,BLK_SIZE);
          send_ReadResponse(response,conn);
          free(response);
          //free(request);
      }else if(trace_type==1)
      {
          //仅仅会接受到来自client的recv任务完成
          WriteRequest* request = (WriteRequest*)malloc(sizeof(WriteRequest));
          memcpy(request,client_conn->recv_region,sizeof(WriteRequest));
          //1.解析接收到的writerequest，将冗余同步情况，requestid，更新的数据块信息等封装存储至map中
          UpdateContext* update_ctx = (UpdateContext*)malloc(sizeof(UpdateContext));
          update_ctx->id=getcurrent_time()*10 +id++;
          update_ctx->requestid = request->requestid;
          update_ctx->conn=conn;
          update_ctx->status=R;
          update_ctx->blockid=request->blockid;
          memcpy(update_ctx->newdata,request->newdata,BLK_SIZE);

          char* key = convert_id_to_string(update_ctx->id);
          cuckoo_insert(update_ctx_map,key,(void*)update_ctx);
          free(key);

          //2.发送同步冗余分块的rdma-write请求
              //2.1根据paritynodeid查找conn，查看len是否超标
              //2.2如果为超，封装信息发送至相应远程内存，修改len；
              //2.3如果超了，通过socket发送更新消息，远程异步update，接受确认update后，len=0；发送信息至远程内存，跟新len
          UpdateEntity* entity = (UpdateEntity*)malloc(sizeof(UpdateEntity));
          memcpy(entity->newdata,request->newdata,BLK_SIZE);
          for (int i = 0; i < R; ++i)
          {
            entity->blockid = request->blockid;
            entity->parityid = request->parityid[i];
            entity->checksum = CHECK_SUM;
            send_UpdateEntity(entity,request->paritynodeid[i],update_ctx->id);
          }
          //3.等待RDMA-WRITE完成，根据id拿到conn，更新数据块然后，发送send给client
          free(entity);
          free(request);
      }else
      {
        printf("error trace_type:%d.\n",trace_type);
      }

    }
    else if (wc->opcode == IBV_WC_SEND){
      printf("Response completed successfully.\n");
    }
    else if (wc->opcode == IBV_WC_RDMA_WRITE)
    {
      //接受rdma-write的完成 获取id 通过id修改updatectx，当status=0时，更新数据块，发送update完成给client_conn
      //printf("1\n");
      char* id = (char*)(uintptr_t)wc->wr_id;
      //printf("2 id:%s\n",id);
      UpdateContext* update_ctx =  (UpdateContext*)cuckoo_get(update_ctx_map,id);
     // printf("3\n");
      update_ctx->status--;
      //printf("4\n");
      if (update_ctx->status==0)
      {
          //printf("5\n");
          memcpy(update_block,update_ctx->newdata,BLK_SIZE);
          WriteResponse* response = (WriteResponse*)malloc(sizeof(WriteResponse));
          response->type=1;
          response->responseid=update_ctx->requestid;
          response->res=1;
          send_WriteResponse(response,update_ctx->conn);
          free(response);
          cuckoo_remove(update_ctx_map,id);
          free(update_ctx);
      }
      free(id);
      //printf("6\n");
    }
    else
      die("on_completion: completion isn't a send or a receive.");

}

void send_UpdateEntity(UpdateEntity* entity,int paritynodeid,long long id)
{
    Connection* conn = datanodes_conns_as_client[paritynodeid];
    //rdma-write
    DataNode_Connection* datanode_conn = (DataNode_Connection*)conn->conn_info;

    struct ibv_send_wr wr, *bad_wr = NULL;
    struct ibv_sge sge;
    memset(&wr, 0, sizeof(wr));
    
    memcpy(datanode_conn->rdma_region,entity,sizeof(UpdateEntity));

    if (conn->len+sizeof(UpdateEntity)>RDMA_BUFFER_SIZE)
    {
      //发送update通知，等待响应后，修改len
      int fd = connect_try(datanodes[paritynodeid].tcp_ip,UPDATE_SOCKET_PORT);

      UpdateMessage* message = (UpdateMessage*)malloc(sizeof(UpdateMessage));
      message->type = 0;
      message->target_region = nodeid;
      send_UpdateMessage(fd,message);
      recv_UpdateMessage(fd,message);
      printf("receive UpdateMessage type=%d \n",message->type);
      if (message->type==1)
      {
          conn->len=0;
          close(fd);  
      }else{
          exit(1);
      }
      free(message);
    }

    char* idstring =  convert_id_to_string(id);
    wr.wr_id = (uintptr_t)idstring;
    wr.opcode = IBV_WR_RDMA_WRITE;
    wr.sg_list = &sge;
    wr.num_sge = 1;
    wr.send_flags = IBV_SEND_SIGNALED;
    wr.wr.rdma.remote_addr = (uintptr_t)datanode_conn->peer_mr.addr+conn->len; //根据len修改addr地址 todo
    wr.wr.rdma.rkey = datanode_conn->peer_mr.rkey;

    sge.addr = (uintptr_t)datanode_conn->rdma_region;
    sge.length = sizeof(UpdateEntity);
    sge.lkey = datanode_conn->rdma_mr->lkey;

    conn->len=conn->len+sizeof(UpdateEntity);//更新conn->len
    TEST_NZ(ibv_post_send(conn->qp, &wr, &bad_wr));

}

void send_WriteResponse(WriteResponse* response,Connection* conn){
    Client_Connection* client_conn = (Client_Connection*)conn->conn_info;

    struct ibv_send_wr wr, *bad_wr = NULL;
    struct ibv_sge sge;
    memset(&wr, 0, sizeof(wr));

    memcpy(client_conn->send_region,response,sizeof(WriteResponse));

    memset(&wr, 0, sizeof(wr));

    wr.wr_id = (uintptr_t)conn;
    wr.opcode = IBV_WR_SEND;
    wr.sg_list = &sge;
    wr.num_sge = 1;
    wr.send_flags = IBV_SEND_SIGNALED;

    sge.addr = (uintptr_t)client_conn->send_region;
    sge.length = sizeof(WriteResponse);
    sge.lkey = client_conn->send_mr->lkey;

    TEST_NZ(ibv_post_send(conn->qp, &wr, &bad_wr));
}

void send_ReadResponse(ReadResponse* response,Connection* conn)
{
   Client_Connection* client_conn = (Client_Connection*)conn->conn_info;

    struct ibv_send_wr wr, *bad_wr = NULL;
    struct ibv_sge sge;
    memset(&wr, 0, sizeof(wr));

    memcpy(client_conn->send_region,response,sizeof(ReadResponse));

    memset(&wr, 0, sizeof(wr));

    wr.wr_id = (uintptr_t)conn;
    wr.opcode = IBV_WR_SEND;
    wr.sg_list = &sge;
    wr.num_sge = 1;
    wr.send_flags = IBV_SEND_SIGNALED;

    sge.addr = (uintptr_t)client_conn->send_region;
    sge.length = sizeof(ReadResponse);
    sge.lkey = client_conn->send_mr->lkey;

    TEST_NZ(ibv_post_send(conn->qp, &wr, &bad_wr));
}

void send_UpdateMessage(int fd,UpdateMessage* message)
{
    size_t needSend =sizeof(UpdateMessage);
    char *sendBuf=(char*)malloc(needSend);
    memcpy(sendBuf,message,needSend); 
    send_bytes(fd,sendBuf,needSend);
    free(sendBuf);
}

void recv_UpdateMessage(int fd,UpdateMessage* message)
{
    size_t needRecv =sizeof(UpdateMessage);
    char *recvBuf=(char *)malloc(needRecv);
    recv_bytes(fd,recvBuf,needRecv);
    memcpy(message,recvBuf,needRecv);  //反序列化
    free(recvBuf);
}



